package com.yujiahao.bigdata.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}

import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util.UUID

object Stream_Source_Kafka {
  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))


    //TODO 4、业务逻辑-- 从kafka中采集数据
    //在流式数据处理中，kafka应用特别多，所以很多的框架都有相应的工具类

    //4.1.定义Kafka参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu", //消费者组名
      //KV反序列化
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //4.2、kafka消费参数
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String]( //这个是读取Kafka的KV类型
        ssc, //上面的环境对象
        LocationStrategies.PreferConsistent, //位置策略，由框架自行选择
        //消费策略
        ConsumerStrategies.Subscribe[String, String](Set("atguigu0819"), kafkaPara))

    //因为kafka中K和分区策略有关系，因此这里数据直接就Values
    val kafkaValuse: DStream[String] = kafkaDStream.map(_.value())
    kafkaValuse.print()

    //TODO 2、启动采集器
    ssc.start()
    //TODO 3、Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
